package org.session.service;

import java.util.Collection;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;

import org.session.model.AddEntryToSessionRequest;
import org.session.model.CreateSessionRequest;
import org.session.model.DeleteKeyFromSessionRequest;
import org.session.model.DeleteSessionRequest;
import org.session.model.EndSessionRequest;
import org.session.model.GetEntriesFromSessionRequest;
import org.session.model.GetEntryFromSessionRequest;
import org.session.model.IRequest;
import org.session.model.ReplaceEntryInSessionRequest;
import org.session.model.SearchSessionRequest;
import org.session.model.Session;
import org.session.model.SessionEntry;
import org.session.util.PropertyLoader;

/**
 * The idea behind this class is to allow for request interception in order to
 * throttle request-push-rate into the repository. Optional Reliable Queueing is
 * used to control message flow.
 * 
 * To keep semantics simple, when queueing is used, methods fall back to
 * returning null values.
 */
public abstract class RequestThrottlerInterceptor implements SessionService {
    private final boolean queueingEnabled;
    private static final long POOL_KILL_WAIT_SECS = 1L;

    // TODO: might want to use a better thread-pool?
    protected ExecutorService executor = Executors.newCachedThreadPool();

    protected abstract Future<Session> createSession(
            CreateSessionRequest request);

    protected abstract Future<Boolean> endSession(EndSessionRequest request);

    protected abstract Future<Boolean> deleteSession(
            DeleteSessionRequest request);

    protected abstract Future<Session> searchSession(
            SearchSessionRequest request);

    protected abstract Future<Boolean> addEntryToSession(
            AddEntryToSessionRequest request);

    protected abstract Future<Boolean> replaceEntryInSession(
            ReplaceEntryInSessionRequest request);

    protected abstract Future<SessionEntry> getEntryFromSession(
            GetEntryFromSessionRequest request);

    protected abstract Future<Map<String, SessionEntry>> getEntriesFromSession(
            GetEntriesFromSessionRequest request);

    protected abstract Future<Boolean> deleteKeyFromSession(
            DeleteKeyFromSessionRequest request);

    {
        queueingEnabled = Boolean.parseBoolean(PropertyLoader.getProperty(
                "service.properties", "queueing.enabled"));
        Runtime.getRuntime().addShutdownHook(new Thread() {
            @Override
            public void run() {
                shutdownRepository(); // assuming this will not block forever
                shutdownPoolAndAwaitTermination();
            }
        });
    }

    @Override
    public Future<Session> createSession(int ttlMillis, boolean idempotence) {
        CreateSessionRequest request = new CreateSessionRequest(ttlMillis,
                idempotence);

        return createSession(request);
    }

    @Override
    public Future<Boolean> endSession(String sessionId, int ttlMillis,
            boolean idempotence) {
        EndSessionRequest request = new EndSessionRequest(sessionId, ttlMillis,
                idempotence);

        return endSession(request);
    }

    @Override
    public Future<Boolean> deleteSession(String sessionId, int ttlMillis,
            boolean idempotence) {
        Future<Boolean> reply = null;
        DeleteSessionRequest request = new DeleteSessionRequest(sessionId,
                ttlMillis, idempotence);

        if (queueingEnabled) {
            enqueue(request);
        } else {
            reply = deleteSession(request);
        }

        return reply;
    }

    @Override
    public Future<Session> searchSession(String sessionId, int ttlMillis,
            boolean idempotence) {
        SearchSessionRequest request = new SearchSessionRequest(sessionId,
                ttlMillis, idempotence);

        return searchSession(request);
    }

    @Override
    public Future<Boolean> addEntryToSession(String sessionId,
            SessionEntry entry, int ttlMillis, boolean idempotence) {
        Future<Boolean> reply = null;
        AddEntryToSessionRequest request = new AddEntryToSessionRequest(
                sessionId, entry, ttlMillis, idempotence);

        if (queueingEnabled) {
            enqueue(request);
        } else {
            reply = addEntryToSession(request);
        }

        return reply;
    }

    @Override
    public Future<Boolean> replaceEntryInSession(String sessionId,
            SessionEntry entry, int ttlMillis, boolean idempotence) {
        Future<Boolean> reply = null;
        ReplaceEntryInSessionRequest request = new ReplaceEntryInSessionRequest(
                sessionId, entry, ttlMillis, idempotence);

        if (queueingEnabled) {
            enqueue(request);
        } else {
            reply = replaceEntryInSession(request);
        }

        return reply;
    }

    @Override
    public Future<SessionEntry> getEntryFromSession(String sessionId,
            String key, int ttlMillis, boolean idempotence) {
        GetEntryFromSessionRequest request = new GetEntryFromSessionRequest(
                sessionId, key, ttlMillis, idempotence);

        return getEntryFromSession(request);
    }

    @Override
    public Future<Map<String, SessionEntry>> getEntriesFromSession(
            String sessionId, Collection<String> keys, int ttlMillis,
            boolean idempotence) {
        GetEntriesFromSessionRequest request = new GetEntriesFromSessionRequest(
                sessionId, keys, ttlMillis, idempotence);

        return getEntriesFromSession(request);
    }

    @Override
    public Future<Boolean> deleteKeyFromSession(String sessionId, String key,
            int ttlMillis, boolean idempotence) {
        Future<Boolean> reply = null;
        DeleteKeyFromSessionRequest request = new DeleteKeyFromSessionRequest(
                sessionId, key, ttlMillis, idempotence);

        if (queueingEnabled) {
            enqueue(request);
        } else {
            reply = deleteKeyFromSession(request);
        }

        return reply;
    }

    private void shutdownPoolAndAwaitTermination() {
        executor.shutdown(); // Disable new tasks from being submitted
        try {
            // Wait a while for existing tasks to terminate
            if (!executor.awaitTermination(POOL_KILL_WAIT_SECS,
                    TimeUnit.SECONDS)) {
                executor.shutdownNow();
                // TODO: log.warn("Wait " + POOL_KILL_WAIT_SECS +
                // " seconds for tasks to respond to being cancelled");
                if (!executor.awaitTermination(POOL_KILL_WAIT_SECS,
                        TimeUnit.SECONDS)) {
                    // TODO: log.error("Thread pool did not terminate");
                }
            }
        } catch (InterruptedException ie) {
            // TODO: log.error("Re-cancel if current thread also interrupted");
            executor.shutdownNow();
            // Preserve interrupt status
            Thread.currentThread().interrupt();
        }
    }

    private void enqueue(final IRequest request) {
        // TODO: implement me
    }

}
